/*
 * MIT License
 *
 * Copyright (c) 2023 北京凯特伟业科技有限公司
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */
package com.je.bpm.engine.impl.asyncexecutor;

import com.je.bpm.engine.ActivitiOptimisticLockingException;
import com.je.bpm.engine.impl.cmd.AcquireTimerJobsCmd;
import com.je.bpm.engine.impl.interceptor.Command;
import com.je.bpm.engine.impl.interceptor.CommandContext;
import com.je.bpm.engine.impl.interceptor.CommandExecutor;
import com.je.bpm.engine.impl.persistence.entity.TimerJobEntity;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.atomic.AtomicBoolean;

/**
 *
 */
public class AcquireTimerJobsRunnable implements Runnable {

    private static Logger log = LoggerFactory.getLogger(AcquireTimerJobsRunnable.class);

    protected final AsyncExecutor asyncExecutor;
    protected final JobManager jobManager;

    protected volatile boolean isInterrupted;
    protected final Object MONITOR = new Object();
    protected final AtomicBoolean isWaiting = new AtomicBoolean(false);

    protected long millisToWait;

    public AcquireTimerJobsRunnable(AsyncExecutor asyncExecutor, JobManager jobManager) {
        this.asyncExecutor = asyncExecutor;
        this.jobManager = jobManager;
    }

    @Override
    public synchronized void run() {
        log.info("{} starting to acquire async jobs due");
        Thread.currentThread().setName("activiti-acquire-timer-jobs");

        final CommandExecutor commandExecutor = asyncExecutor.getProcessEngineConfiguration().getCommandExecutor();

        while (!isInterrupted) {
            try {
                final AcquiredTimerJobEntities acquiredJobs = commandExecutor.execute(new AcquireTimerJobsCmd(asyncExecutor));
                commandExecutor.execute(new Command<Void>() {

                    @Override
                    public Void execute(CommandContext commandContext) {
                        for (TimerJobEntity job : acquiredJobs.getJobs()) {
                            jobManager.moveTimerJobToExecutableJob(job);
                        }
                        return null;
                    }
                });

                // if all jobs were executed
                millisToWait = asyncExecutor.getDefaultTimerJobAcquireWaitTimeInMillis();
                int jobsAcquired = acquiredJobs.size();
                if (jobsAcquired >= asyncExecutor.getMaxTimerJobsPerAcquisition()) {
                    millisToWait = 0;
                }
            } catch (ActivitiOptimisticLockingException optimisticLockingException) {
                if (log.isDebugEnabled()) {
                    log.debug("Optimistic locking exception during timer job acquisition. If you have multiple timer executors running against the same database, "
                                    + "this exception means that this thread tried to acquire a timer job, which already was acquired by another timer executor acquisition thread."
                                    + "This is expected behavior in a clustered environment. "
                                    + "You can ignore this message if you indeed have multiple timer executor acquisition threads running against the same database. " + "Exception message: {}",
                            optimisticLockingException.getMessage());
                }
            } catch (Throwable e) {
                log.error("exception during timer job acquisition: {}", e.getMessage(), e);
                millisToWait = asyncExecutor.getDefaultTimerJobAcquireWaitTimeInMillis();
            }

            if (millisToWait > 0) {
                try {
                    if (log.isDebugEnabled()) {
                        log.debug("timer job acquisition thread sleeping for {} millis", millisToWait);
                    }
                    synchronized (MONITOR) {
                        if (!isInterrupted) {
                            isWaiting.set(true);
                            MONITOR.wait(millisToWait);
                        }
                    }

                    if (log.isDebugEnabled()) {
                        log.debug("timer job acquisition thread woke up");
                    }
                } catch (InterruptedException e) {
                    if (log.isDebugEnabled()) {
                        log.debug("timer job acquisition wait interrupted");
                    }
                } finally {
                    isWaiting.set(false);
                }
            }
        }

        log.info("{} stopped async job due acquisition");
    }

    public void stop() {
        synchronized (MONITOR) {
            isInterrupted = true;
            if (isWaiting.compareAndSet(true, false)) {
                MONITOR.notifyAll();
            }
        }
    }

    public long getMillisToWait() {
        return millisToWait;
    }

    public void setMillisToWait(long millisToWait) {
        this.millisToWait = millisToWait;
    }
}
